package drds.binlog.parse.mysql.dbsync;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.database_driver.packets.server.FieldPacket;
import drds.binlog.database_driver.packets.server.ResultSetPacket;
import drds.binlog.parse.FieldMetaData;
import drds.binlog.parse.TableMetaData;
import drds.binlog.parse.exception.ParseException;
import drds.binlog.parse.mysql.Connection;
import drds.binlog.parse.mysql.ddl.FastsqlDdlParser;
import drds.binlog.parse.mysql.table_meta_data.BaseOnDatabaseTableMetaDataGetter;
import drds.binlog.parse.mysql.table_meta_data.BaseOnMemoryTableMetaDataGetter;
import drds.binlog.parse.mysql.table_meta_data.TableMetaDataGetter;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 处理table meta解析和缓存
 */
@Author(name = Authors.LI_YANG)
public class TableMetaDataCache
{

    public static final String column_name = "column_name";
    public static final String column_type = "column_type";
    public static final String is_nullable = "is_nullable";
    public static final String column_key = "column_key";
    public static final String column_default = "column_default";
    public static final String extra = "extra";
    //
    @Setter
    @Getter
    private Connection connection;
    @Setter
    @Getter
    private boolean isOnRDS = false;
    @Setter
    @Getter
    private TableMetaDataGetter tableMetaDataGetter;
    // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表
    @Setter
    @Getter
    private LoadingCache<String, TableMetaData> schemaNameAndTableNameToTableMetaDataCache;

    public TableMetaDataCache(Connection connection, TableMetaDataGetter tableMetaDataGetter)
    {
        this.connection = connection;
        this.tableMetaDataGetter = tableMetaDataGetter;
        // 如果持久存储的表结构为空，从db里面获取下
        if (tableMetaDataGetter == null)
        {
            this.schemaNameAndTableNameToTableMetaDataCache = CacheBuilder.newBuilder().build(new CacheLoader<String, TableMetaData>()
            {

                @Override
                public TableMetaData load(String schemaNameAndTableName) throws Exception
                {
                    try
                    {
                        return getTableMetaData(schemaNameAndTableName);
                    } catch (Throwable e)
                    {
                        // 尝试做一次retry操作
                        try
                        {
                            TableMetaDataCache.this.connection.reconnect();
                            return getTableMetaData(schemaNameAndTableName);
                        } catch (IOException e1)
                        {
                            throw new ParseException("fetch failed by table metadata:" + schemaNameAndTableName, e1);
                        }
                    }
                }

            });
        }

        try
        {
            ResultSetPacket resultSetPacket = this.connection.query("show global variables  like 'rds\\_%'");
            if (resultSetPacket.getFieldValueList().size() > 0)
            {
                isOnRDS = true;
            }
        } catch (IOException e)
        {
        }
    }

    public static List<FieldMetaData> parseTableMeta(String schemaName, String tableName, ResultSetPacket packet)
    {
        if (packet.getFieldValueList().size() > 1)
        {
            String createDDL = packet.getFieldValueList().get(1);
            BaseOnMemoryTableMetaDataGetter baseOnMemoryTableMetaDataGetter = new BaseOnMemoryTableMetaDataGetter();
            baseOnMemoryTableMetaDataGetter.apply(BaseOnDatabaseTableMetaDataGetter.INIT_POSITION, schemaName, createDDL, null);
            TableMetaData tableMetaData = baseOnMemoryTableMetaDataGetter.find(schemaName, tableName);
            return tableMetaData.getFieldMetaDataList();
        } else
        {
            return new ArrayList<FieldMetaData>();
        }
    }

    /**
     * 处理desc table的结果
     */
    public static List<FieldMetaData> parseTableMetaByDesc(ResultSetPacket resultSetPacket)
    {
        Map<String, Integer> nameMaps = new HashMap<String, Integer>(6, 1f);
        int index = 0;
        for (FieldPacket fieldPacket : resultSetPacket.getFieldDescriptorList())
        {
            nameMaps.put(fieldPacket.getOriginalColumnName(), index++);
        }

        int size = resultSetPacket.getFieldDescriptorList().size();
        int count = resultSetPacket.getFieldValueList().size() / resultSetPacket.getFieldDescriptorList().size();
        List<FieldMetaData> fieldMetaDataList = new ArrayList<FieldMetaData>();
        for (int i = 0; i < count; i++)
        {
            FieldMetaData fieldMetaData = new FieldMetaData();
            // 做一个优化，使用String.intern()，共享String对象，减少内存使用
            fieldMetaData.setColumnName(resultSetPacket.getFieldValueList().get(nameMaps.get(column_name) + i * size).intern());
            fieldMetaData.setColumnType(resultSetPacket.getFieldValueList().get(nameMaps.get(column_type) + i * size));
            fieldMetaData.setNullable(StringUtils.equalsIgnoreCase(resultSetPacket.getFieldValueList().get(nameMaps.get(is_nullable) + i
                            * size),
                    "YES"));
            fieldMetaData.setPrimaryKey("PRI".equalsIgnoreCase(resultSetPacket.getFieldValueList().get(nameMaps.get(column_key) + i * size)));
            fieldMetaData.setUniqueKey("UNI".equalsIgnoreCase(resultSetPacket.getFieldValueList().get(nameMaps.get(column_key) + i * size)));
            // 特殊处理引号
            fieldMetaData.setDefaultValue(FastsqlDdlParser.unescapeQuotaName(resultSetPacket.getFieldValueList()
                    .get(nameMaps.get(column_default) + i * size)));
            fieldMetaData.setExtra(resultSetPacket.getFieldValueList().get(nameMaps.get(extra) + i * size));

            fieldMetaDataList.add(fieldMetaData);
        }

        return fieldMetaDataList;
    }

    /**
     * 查询表结构  show create table
     */
    private TableMetaData getTableMetaData(String schemaNameAndTableName) throws IOException
    {
        try
        {
            ResultSetPacket resultSetPacket = connection.query("show create table " + schemaNameAndTableName);
            String[] names = StringUtils.split(schemaNameAndTableName, "`.`");
            String schema = names[0];
            String table = names[1].substring(0, names[1].length());
            return new TableMetaData(schema, table, parseTableMeta(schema, table, resultSetPacket));
        } catch (Throwable e)
        { // fallback to desc table
            ResultSetPacket resultSetPacket = connection.query("desc " + schemaNameAndTableName);
            String[] names = StringUtils.split(schemaNameAndTableName, "`.`");
            String schema = names[0];
            String table = names[1].substring(0, names[1].length());
            return new TableMetaData(schema, table, parseTableMetaByDesc(resultSetPacket));
        }
    }


    public TableMetaData getTableMetaData(String schemaName, String tableName, boolean useCache, EntryPosition entryPosition)
    {
        TableMetaData tableMetaData = null;
        if (tableMetaDataGetter != null)
        {
            tableMetaData = tableMetaDataGetter.find(schemaName, tableName);
            if (tableMetaData == null)
            {
                // 因为条件变化，可能第一次的tableMeta没取到，需要从db获取一次，并记录到snapshot中
                String schemaNameAndTableName = getSchemaNameAndTableName(schemaName, tableName);
                try
                {
                    ResultSetPacket resultSetPacket = connection.query("show create table " + schemaNameAndTableName);
                    String createTableDdl = null;
                    if (resultSetPacket.getFieldValueList().size() > 0)
                    {
                        createTableDdl = resultSetPacket.getFieldValueList().get(1);
                    }
                    // 强制覆盖掉内存值
                    tableMetaDataGetter.apply(entryPosition, schemaName, createTableDdl, "first");
                    tableMetaData = tableMetaDataGetter.find(schemaName, tableName);
                } catch (IOException e)
                {
                    throw new ParseException("fetch failed by table metadata:" + schemaNameAndTableName, e);
                }
            }
            return tableMetaData;
        } else
        {
            if (!useCache)
            {
                schemaNameAndTableNameToTableMetaDataCache.invalidate(getSchemaNameAndTableName(schemaName, tableName));
            }

            return schemaNameAndTableNameToTableMetaDataCache.getUnchecked(getSchemaNameAndTableName(schemaName, tableName));
        }
    }

    public void clearTableMeta(String schemaName, String tableName)
    {
        if (tableMetaDataGetter != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            schemaNameAndTableNameToTableMetaDataCache.invalidate(getSchemaNameAndTableName(schemaName, tableName));
        }
    }

    public void clearTableMetaWithSchemaName(String schemaName)
    {
        if (tableMetaDataGetter != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            for (String schemaNameAndTableName : schemaNameAndTableNameToTableMetaDataCache.asMap().keySet())
            {
                if (StringUtils.startsWithIgnoreCase(schemaNameAndTableName, schemaName + "."))
                {
                    // removeNames.add(name);
                    schemaNameAndTableNameToTableMetaDataCache.invalidate(schemaNameAndTableName);
                }
            }
        }
    }

    public void clearTableMeta()
    {
        if (tableMetaDataGetter != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            schemaNameAndTableNameToTableMetaDataCache.invalidateAll();
        }
    }

    /**
     * 更新一下本地的表结构内存
     */
    public boolean apply(EntryPosition entryPosition, String schemaName, String ddl, String extra)
    {
        if (tableMetaDataGetter != null)
        {
            return tableMetaDataGetter.apply(entryPosition, schemaName, ddl, extra);
        } else
        {
            // ignore
            return true;
        }
    }

    private String getSchemaNameAndTableName(String schemaName, String tableName)
    {
        StringBuilder sb = new StringBuilder();
        return sb.append('`')
                .append(schemaName)
                .append('`')
                .append('.')
                .append('`')
                .append(tableName)
                .append('`')
                .toString();
    }


}
